package com.zhg.shortlink.service;

import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

/**
 * 基于时间分片的布隆过滤器服务
 * 解决过期数据无法删除的问题
 * @author 朱洪刚
 * @version 1.0
 * @data 2025/10/22 14:02
 */
@Slf4j
@RequiredArgsConstructor
@Service
public class TimeBasedBloomFilterService {

    private final RedisBloomFilterService redisBloomFilterService;
    private final BloomFilterStreamService streamService;
    private final RedissonClient redissonClient; // 新增

    @Value("${shortlink.bloom.time-slice.hours:6}")
    private int timeSliceHours; // 时间分片粒度（小时）


    @Value("${shortlink.bloom.time-slice.local-keep-count:8}")
    private int localKeepSliceCount; // 本地保留的时间片数量（2天=6*8）

    // 时间分片布隆过滤器映射
    private final ConcurrentMap<String, TimeSliceBloomFilter> timeSlices = new ConcurrentHashMap<>();

    // 当前活跃的时间片
    private volatile String currentTimeSlice;

    @PostConstruct
    public void init() {
        // 初始化当前时间片
        currentTimeSlice = getCurrentTimeSliceKey();
        createTimeSlice(currentTimeSlice);

        log.info("时间分片布隆过滤器初始化完成，当前时间片: {}", currentTimeSlice);
    }

    /**
     * 分层检查：时间分片 + 本地 + Redis
     */
    public boolean mightContain(String shortCode) {
        // 第一层：检查所有活跃的时间分片
        for (TimeSliceBloomFilter slice : timeSlices.values()) {
            if (slice.mightContain(shortCode)) {
                return true;
            }
        }

        // 第二层：检查Redis布隆过滤器（兜底）
        return redisBloomFilterService.mightContain(shortCode);
    }

    /**
     * 添加到当前时间片
     */
    public void add(String shortCode) {
        String currentSlice = getCurrentTimeSliceKey();

        // 如果时间片发生变化，创建新的时间片
        if (!currentSlice.equals(currentTimeSlice)) {
            synchronized (this) {
                if (!currentSlice.equals(currentTimeSlice)) {
                    createTimeSlice(currentSlice);
                    currentTimeSlice = currentSlice;
                    log.info("创建新的时间片: {}", currentSlice);
                }
            }
        }

        // 添加到当前时间片
        TimeSliceBloomFilter slice = timeSlices.get(currentTimeSlice);
        if (slice != null) {
            slice.add(shortCode);
        }

        // 同时添加到Redis（全局共享）
        redisBloomFilterService.addAsync(shortCode);

        // 发布到Stream同步到其他节点
        streamService.publishNewShortCode(shortCode);
    }

    /**
     * 定时清理过期的时间片
     */
    @Scheduled(fixedRate = 300000) // 每5分钟检查一次
    public void cleanupExpiredSlices() {
        String lockKey = "bloom_filter_cleanup_lock";
        RLock lock = redissonClient.getLock(lockKey);

        try {
            // 尝试获取锁，最多等待10秒，锁定60秒
            if (lock.tryLock(10, 60, TimeUnit.SECONDS)) {
                try {
                    log.info("获取清理锁成功，开始清理过期时间片 - 节点: {}", getNodeId());
                    doCleanupExpiredSlices();
                } finally {
                    lock.unlock();
                    log.info("清理完成，释放锁 - 节点: {}", getNodeId());
                }
            } else {
                log.debug("未能获取清理锁，跳过本次清理 - 节点: {}", getNodeId());
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("清理任务被中断 - 节点: {}", getNodeId());
        } catch (Exception e) {
            log.error("清理过程中发生异常 - 节点: {}", getNodeId(), e);
        }
    }

    /**
     * 实际的清理逻辑
     */
    private void doCleanupExpiredSlices() {
        try {
            List<String> expiredSlices = new ArrayList<>();

            for (String sliceKey : timeSlices.keySet()) {
                if (isSliceExpired(sliceKey)) {
                    expiredSlices.add(sliceKey);
                }
            }

            for (String expiredSlice : expiredSlices) {
                TimeSliceBloomFilter removed = timeSlices.remove(expiredSlice);
                if (removed != null) {
                    log.info("清理过期时间片: {}, 元素数量: {}, 执行节点: {}",
                            expiredSlice, removed.getApproximateElementCount(), getNodeId());
                }
            }

            log.info("时间片清理完成，当前活跃片数: {}, 执行节点: {}", timeSlices.size(), getNodeId());

        } catch (Exception e) {
            log.error("清理过期时间片失败 - 节点: {}", getNodeId(), e);
        }
    }

    /**
     * 获取节点ID
     */
    private String getNodeId() {
        return streamService != null ? streamService.getNodeId() : "unknown";
    }

    private String getCurrentTimeSliceKey() {
        LocalDateTime now = LocalDateTime.now();
        // 按小时分片，向下取整
        LocalDateTime sliceTime = now.withMinute(0).withSecond(0).withNano(0)
                .withHour((now.getHour() / timeSliceHours) * timeSliceHours);
        return sliceTime.format(DateTimeFormatter.ofPattern("yyyyMMdd_HH"));
    }

    private void createTimeSlice(String sliceKey) {
        TimeSliceBloomFilter slice = new TimeSliceBloomFilter(sliceKey);
        timeSlices.put(sliceKey, slice);
    }

    private boolean isSliceExpired(String sliceKey) {
        try {
            LocalDateTime sliceTime = LocalDateTime.parse(sliceKey,
                    DateTimeFormatter.ofPattern("yyyyMMdd_HH"));
            LocalDateTime expireTime = sliceTime.plusHours(timeSliceHours * localKeepSliceCount);
            return LocalDateTime.now().isAfter(expireTime);
        } catch (Exception e) {
            log.warn("解析时间片key失败: {}", sliceKey);
            return true; // 解析失败认为已过期
        }
    }

    public String getStats() {
        return String.format("时间分片统计 - 活跃片数: %d, 当前片: %s, 总元素数: %d",
                timeSlices.size(), currentTimeSlice,
                timeSlices.values().stream().mapToLong(TimeSliceBloomFilter::getApproximateElementCount).sum());
    }

}
